package com.acdc.mqtt.service;

import org.eclipse.paho.client.mqttv3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import com.acdc.mqtt.entity.MqttMsg;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.annotation.JsonInclude;

/**
 * MQTT生产者服务
 */
@Service
public class MqttProducerService {
    
    private static final Logger log = LoggerFactory.getLogger(MqttProducerService.class);
    
    @Autowired
    @Qualifier("producerMqttClient")
    private MqttClient mqttClient;
    
    @Autowired
    @Qualifier("producerMqttConnectOptions")
    private MqttConnectOptions connectOptions;
    
    // 配置ObjectMapper，忽略null值属性
    private final ObjectMapper objectMapper = new ObjectMapper()
        .setSerializationInclusion(JsonInclude.Include.NON_NULL)
        .configure(com.fasterxml.jackson.databind.SerializationFeature.WRITE_ENUMS_USING_TO_STRING, true);
    
    @PostConstruct
    public void init() {
        try {
            // 设置回调
            mqttClient.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    log.error("MQTT连接丢失", cause);
                }
                
                @Override
                public void messageArrived(String topic, MqttMessage message) {
                    log.info("收到消息: topic={}, message={}", topic, new String(message.getPayload()));
                }
                
                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    log.info("消息发送完成");
                }
            });
            
            // 连接到MQTT代理
            mqttClient.connect(connectOptions);
            log.info("MQTT生产者服务初始化成功");
            
        } catch (MqttException e) {
            log.error("MQTT生产者服务初始化失败", e);
        }
    }
    
    /**
     * 发送消息
     * @param topic 主题
     * @param message 消息内容
     * @param qos 服务质量
     */
    public void publishMessage(String topic, MqttMsg message, int qos) {
        try {
            if (mqttClient != null && mqttClient.isConnected()) {
                // 将MqttMsg转换为JSON字符串，忽略null值属性，并去除换行符、回车符、制表符
                String json = objectMapper.writeValueAsString(message);
                json = json.replaceAll("\n", "").replaceAll("\r", "").replaceAll("\t", "");
                MqttMessage mqttMessage = new MqttMessage(json.getBytes());
                mqttMessage.setQos(qos);
                mqttClient.publish(topic, mqttMessage);
                log.info("消息发送成功: topic={}, message={}, qos={}", topic, message, qos);
            } else {
                log.warn("MQTT客户端未连接，无法发送消息");
            }
        } catch (MqttException e) {
            log.error("发送消息失败: topic={}, message={}", topic, message, e);
        } catch (JsonProcessingException e) {
            log.error("转换消息为JSON失败: topic={}, message={}", topic, message, e);
        }
    }
    
    /**
     * 发送消息（默认QoS=1）
     * @param topic 主题
     * @param message 消息内容
     */
    public void publishMessage(String topic, MqttMsg message) {
        publishMessage(topic, message, 1);
    }
    
    /**
     * 检查连接状态
     * @return 是否已连接
     */
    public boolean isConnected() {
        return mqttClient != null && mqttClient.isConnected();
    }
    
    @PreDestroy
    public void destroy() {
        try {
            if (mqttClient != null && mqttClient.isConnected()) {
                mqttClient.disconnect();
                mqttClient.close();
                log.info("MQTT生产者服务已关闭");
            }
        } catch (MqttException e) {
            log.error("关闭MQTT生产者服务失败", e);
        }
    }
} 